package org.shj.spark.dataframe;

import org.apache.spark.api.java.function.ForeachFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

public class SparkSQLTest {

	public static void main(String[] args) {
		SparkSession ss = SparkSession.builder().appName("SparkSQLTest").master("local").getOrCreate();
		//ss.sparkContext().setLogLevel("WARN");
		
		Dataset<Row> json = ss.read().json("src/main/resources/employees.json");
		
		json.createOrReplaceTempView("emp");
		
		Dataset<Row> sql = ss.sql("select * from emp where salary > 3500");
		sql.foreach(new ForeachFunction<Row>() {
			private static final long serialVersionUID = -6524717773878953798L;

			@Override
			public void call(Row r) throws Exception {
				System.out.println(r);				
			}
		});
		
		
		/**
		 * save 方法指定的是目录。另外，在pom里加入hadoop的包后，会认为是要存在HDFS上，所以会报错。
		 * 运行在本地测时，需要去掉Hadoop的包
		 * 下面append模式，并不是在上一次的结果文件上追加内容，而是另外生成一个文件
		 * 如果是 overwrite 模式的话，则会清除上一次的结果文件
		 * 
		 */
		json.filter("salary > 3000").write().format("json")
			.mode(SaveMode.Append).save("E:/workspace/scala/sparkjava/output/emp");
		
		
		System.out.println("===============");
		//读取 parquet 格式的文件
		Dataset<Row> parquetDs = ss.read().parquet("src/main/resources/users.parquet");
		parquetDs.foreach(new ForeachFunction<Row>() {
			private static final long serialVersionUID = -6524717773878953798L;

			@Override
			public void call(Row r) throws Exception {
				System.out.println(r);				
			}
		});
		parquetDs.write().mode(SaveMode.Overwrite).save("E:/workspace/scala/sparkjava/output/parquet");
		
		ss.stop();
	}

}
